/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.data.a2b.server;

import cn.ac.iie.ulss.data.a2b.commons.RuntimeEnv;
import cn.ac.iie.ulss.data.a2b.config.Configuration;
import cn.ac.iie.ulss.data.a2b.rocketmq.RocketMQConsumer;
import static cn.ac.iie.ulss.data.a2b.rocketmq.RocketMQConsumer.consumer;
import cn.ac.iie.ulss.data.a2b.rocketmq.RocketMQProducer;
import static cn.ac.iie.ulss.data.a2b.rocketmq.RocketMQProducer.producer;
import cn.ac.iie.ulss.data.a2b.worker.Worker;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.avro.Protocol;
import org.apache.avro.generic.GenericRecord;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

/**
 *
 * @author alexmu
 */
public class Dataa2bStartUp {

    static Logger logger = null;
    public static LinkedBlockingQueue<GenericRecord> a2bLinkedBlockingQueue = new LinkedBlockingQueue(300000);
    public static Protocol protocol = null;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = Logger.getLogger(Dataa2bStartUp.class.getName());
    }

    public static void showUsage() {
        System.out.println("Usage:java -jar ");
    }

    public static void main(String[] args) {
        try {
            init();
            startup();
        } catch (Exception ex) {
            logger.error("starting data a2b is failed for " + ex.getMessage(), ex);
        }
        System.exit(0);
    }

    private static void startup() throws Exception {
        logger.info("starting data a2b...");

        RocketMQProducer mQProducer = new RocketMQProducer();
        mQProducer.init(RuntimeEnv.getParamAsString(RuntimeEnv.DES_GROUP), RuntimeEnv.getParamAsString(RuntimeEnv.DES_NAMESERVER));

        for (int i = 0; i < RuntimeEnv.getParamAsInteger(RuntimeEnv.WORKER_SIZE); i++) {
            Worker worker = new Worker();
            worker.init();
            Thread senThread = new Thread(worker);
            senThread.setName(RuntimeEnv.getParamAsString(RuntimeEnv.SRC_TOPIC) + "-" + RuntimeEnv.getParamAsString(RuntimeEnv.DES_TOPIC) + "-" + i);
            senThread.start();
        }

        RocketMQConsumer mQConsumer = new RocketMQConsumer();
        mQConsumer.startConsumer(RuntimeEnv.getParamAsString(RuntimeEnv.SRC_TOPIC));

        logger.info("start data a2b successfully");

        while (true) {
            try {
                Thread.sleep(5000L);
            } catch (Exception e) {
            }
            logger.info(RuntimeEnv.getParamAsString(RuntimeEnv.SRC_TOPIC) + " to "
                    + RuntimeEnv.getParamAsString(RuntimeEnv.DES_TOPIC)
                    + " LinkedBlockingQueue size is " + a2bLinkedBlockingQueue.size());
        }
    }

    private static void init() throws Exception {
        String configurationFileName = "ulss-data-a2b.properties";
        logger.info("initializing data a2b...");
        logger.info("getting configuration from configuration file " + configurationFileName);
        Configuration conf = Configuration.getConfiguration(configurationFileName);
        if (conf == null) {
            throw new Exception("reading " + configurationFileName + " is failed.");
        }

        logger.info("initializng runtime enviroment...");
        if (!RuntimeEnv.initialize(conf)) {
            throw new Exception("initializng runtime enviroment is failed");
        }
        logger.info("initialize runtime enviroment successfully");

        logger.info("register shutdown hook");
        KillHandler killHandler = new KillHandler();
        killHandler.registerSignal("TERM");
        logger.info("register shutdown hook successfully");
        logger.info("intialize data dispatch server successfully");
    }

    public static void stop() {
        try {
            consumer.shutdown();
            try {
                Thread.sleep(20000L);
            } catch (Exception e) {
            }
            while (!a2bLinkedBlockingQueue.isEmpty()) {
                try {
                    Thread.sleep(2000L);
                } catch (Exception e) {
                }
                logger.info(RuntimeEnv.getParamAsString(RuntimeEnv.SRC_TOPIC) + " to "
                        + RuntimeEnv.getParamAsString(RuntimeEnv.DES_TOPIC)
                        + " LinkedBlockingQueue is not empty and size is " + a2bLinkedBlockingQueue.size());
            }
            logger.info(RuntimeEnv.getParamAsString(RuntimeEnv.SRC_TOPIC) + " to "
                    + RuntimeEnv.getParamAsString(RuntimeEnv.DES_TOPIC)
                    + " LinkedBlockingQueue is empty");
            producer.shutdown();
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
        logger.info("data a2b has beed stopped");
        System.exit(0);
    }
}
